package a_aa_amainpackage.a_o2odata_deal.utils

import a_aa_amainpackage.a_o2odata_deal.config.config._
import com.alibaba.fastjson.JSON
import org.apache.spark.sql.{DataFrame, SQLContext}
object tm_shop {

  val outpath = s"${shopres}"

  def tm_shop(sqlContext:SQLContext,table:DataFrame):Unit = {
    table.registerTempTable("tmTable")


    val sql1 =
      """
        |select shopId,cast(sum(sellCount) as bigint) as totalSellCount,
        |cast(sum(salesAmount) as Double) as totalSalesAmount from tmTable
        |group by shopId
      """.stripMargin

    val res1 =sqlContext.sql(sql1)
    res1.registerTempTable("ResTable")

    val sql3 =
      """
        |select
        |a.is_flagship,
        |b.totalSellCount,
        |b.totalSalesAmount,
        |a.shopId,
        |a.shopName,
        |a.creditLevel,
        |a.starts,
        |a.evaluates,
        |a.fans,
        |a.is_oversea,
        |a.platformId,
        |a.platformName,
        |a.shopType,
        |a.shopUrl,
        |a.address,
        |a.administrative_region,
        |a.city,
        |a.city_grade,
        |a.city_origin,
        |a.district,
        |a.district_origin,
        |a.economic_division,
        |a.if_city,
        |a.if_district,
        |a.if_state_level_new_areas,
        |a.name,
        |a.poor_counties,
        |a.province,
        |a.regional_ID,
        |a.rural_demonstration_counties,
        |a.rural_ecommerce,
        |a.the_belt_and_road_city ,
        |a.the_belt_and_road_province ,
        |a.the_yangtze_river_economic_zone_city,
        |a.the_yangtze_river_economic_zone_province,
        |a.urban_agglomerations,
        |a.longitude,
        |a.latitude,
        |a.is_market,
        |a.is_o2o,
        |a.aedzId,
        |a.town,
        |a.timeStamp,
        |a.registration_institution,
        |a.userId
        |from  ResTable b left join tmTable a on a.shopId=b.shopId
       """.stripMargin
    val res3 = sqlContext.sql(sql3).dropDuplicates("shopId")

    res3.show()

    res3.createOrReplaceTempView("res3")
    ////////////////////////////////
    val res3_v1 =  sqlContext.sql(
      """
        |select
        |*,
        |case when province in ('山东省','北京市') and city in ('济南市','莱芜市','大兴区') then (case
        |when city in ('济南市','莱芜市') then '济南市' when city  ='大兴区' then '北京市' else city end) else city end as city_v1,
        |case when province in ('山东省','浙江省') and city in ('济南市','莱芜市','宁波市') and district in ('莱城区','市辖区','江东区') then
        |(case when district in ('莱城区','市辖区') then '莱芜区' when district = '江东区' then '鄞州区'
        |else district end) else district end as district_v1,
        |case when province in ('山东省','浙江省') and city in ('济南市','莱芜市','宁波市') then (case when district in ('市辖区','莱城区') then '371202'
        |when district = '钢城区' then '371203' when regional_ID='370117' then '371203' when regional_ID='330204' then '330212'
        |else regional_ID end) else regional_ID end as regional_ID_v1
        |from
        |res3
      """.stripMargin).drop("city","district","regional_ID")
      .withColumnRenamed("city_v1","city")
      .withColumnRenamed("district_v1","district")
      .withColumnRenamed("regional_ID_v1","regional_ID")
      .dropDuplicates("shopId")

    //////////////////////////

    val value = res3_v1.toJSON.rdd.map(line => {
      val nObject = JSON.parseObject(line)
      val evaluates = nObject.getOrDefault("evaluates", "-1").toString
      nObject.put("evaluates", evaluates)
      nObject.toJSONString
    })
    //临时注释  需要放开
    sqlContext.read.json(value).repartition(4).write.orc(s"s3a://dws-data/g_shop/${year}/${month}/${platform}")
    //双十一代码
    //sqlContext.read.json(value).repartition(5).write.orc(s"s3a://dws-data/g_shop/2019/shuang11/normalSales_finally/tmall")
    sqlContext.read.json(value).repartition(4).write.json(s"${outpath}")
  }

}
